/*
 * DataQueue.cpp
 *
 *  Created on: 2013-9-5
 *      Author: root
 */

#include "DataQueue.h"
//#include <glog/logging.h>
#include <zmq.h>
#include <zmq_utils.h>

namespace core {
namespace zmq {

DataRecvThread::DataRecvThread(){
	queueName = "";
	ctx = NULL;
	sock = NULL;
	runing = false;
	isExist = false;
}

int DataRecvThread::Init(std::string queueName, int highWater, int mode){
	this->queueName = queueName;

    ctx = zmq_init (1);
    assert (ctx);
    sock = zmq_socket (ctx, mode);
    assert (sock);
    zmq_setsockopt (sock, ZMQ_RCVHWM, &highWater, sizeof (highWater));

    zmq_bind (sock, this->queueName.c_str());

	int linger = 0;
    zmq_setsockopt (sock, ZMQ_LINGER, &linger, sizeof (linger));

	runing = false;
	this->isExist = true;


	return 1;
}

DataRecvThread::DataRecvThread(std::string queueName, int highWater, int mode){
	Init(queueName, highWater, mode);
}

DataRecvThread::~DataRecvThread() {
	runing = false; //通知线程结束的处理
	try {
		zmq_unbind (sock, this->queueName.c_str());
	    // Let events some time
	    sleep (1);

		//Assertion failed: pfd.revents & POLLIN
		//LOG(INFO)<<"stop 0";
		zmq_close (sock);
		//LOG(INFO)<<"stop 1";
		if(!this->isExist){
			//LOG(INFO)<<"stop.";
			usleep(100);
		}

		zmq_term (ctx);
		//LOG(INFO)<<"stop 2";
	} catch(...)
	{
	}

	if(!this->isExist){
		//LOG(INFO)<<"stop...";
		usleep(100);
	}
	//LOG(INFO)<<"stop";
}

bool DataRecvThread::Start() {
	if(runing) return true;

	runing = true;
	this->thrPull = boost::thread(boost::ref(*this));
	return true;
}

void DataRecvThread::operator()() {
	//LOG(INFO)<< "ready";
	this->isExist = false;
	while (runing && sock!= NULL) {
		try {
			oneWork();
		} catch (std::exception const &e) {
	        //LOG(ERROR)<< "DataRecvThread():" << e.what();
			//LOG(ERROR)<< "stop error";
	    }catch (...) {
			//LOG(ERROR)<< "stop error";
		}
	}

	//LOG(INFO)<< "quit";
	this->isExist = true;
}

void DataRecvThread::WaitForDone(){
    //LOG(INFO) << "WaitForDone";
	this->thrPull.join();
}

size_t DataRecvThread::recvBuf(void *buf_, size_t len_, int flags_ )
{
    int nbytes = 0;
    int error = 0;
    do { //try again 再次尝试
    	nbytes = zmq_recv (sock, buf_, len_, flags_);
    } while (nbytes < 0 && EAGAIN == (error = zmq_errno()) );

    if (nbytes >= 0) { //正常数据
        return (size_t) nbytes;
    }

    //LOG(WARNING) << "\n[ \033[33mWARNING\033[0m ] recvBuf - error:" << error;

    return 0; // throw error_t ();
}

void DataRecvThread::recvC(const boost::uint8_t &c) {
	recvBuf((void*) &c, sizeof(c));
}

void DataRecvThread::recvS(const boost::uint16_t &s) {
	recvBuf((void*) &s, sizeof(s));
}

void DataRecvThread::recvI(const boost::int32_t &i) {
	recvBuf((void*) &i, sizeof(i));
}

void DataRecvThread::recvL(const boost::int64_t &l) {
	recvBuf((void*) &l, sizeof(l));
}

void DataRecvThread::recv(string &str) {
	//sock->recv((void*) buf, MAX_TEXT);
	int i = recvBuf((void*) buf, MAX_TEXT);
	buf[i] = 0;
	//如果有'\0'数据截断
	//str = buf;
	str.clear();
	str.append(buf,i);
}

/*
void DataRecvThread::recvV(vector<std::string> &array) {
	boost::uint16_t size;
	recvS(size);

	for (boost::uint16_t i = 0; i < size; i++) {
		std::string s;
		recv(s);
		array.push_back(s);
	}
}
*/

/////////////////////////////
DataSend::DataSend(){
	ctx = NULL;
	sock = NULL;
	queueName = "";

	isDrop = false;

}

DataSend::DataSend(std::string queueName, int highWater, int mode){
	this->queueName = queueName;
    ctx = zmq_init (1);
    assert (ctx);
    sock = zmq_socket (ctx, mode);
    assert (sock);

    zmq_setsockopt (sock, ZMQ_SNDHWM, &highWater, sizeof (highWater));

	zmq_connect(sock, this->queueName.c_str());

	int linger = 0;
    zmq_setsockopt (sock, ZMQ_LINGER, &linger, sizeof (linger));

    isDrop = false;

}
DataSend::~DataSend() {
	zmq_disconnect(sock, this->queueName.c_str());
	////LOG(INFO)<<"snd stop 0";
	zmq_close (sock);
	////LOG(INFO)<<"snd stop 1";
	zmq_term (ctx);
	////LOG(INFO)<<"snd stop 2";
}

void DataSend::sendC(const boost::uint8_t &c, int flag) {
	sendBuf((void*) &c, sizeof(c), flag);
}

void DataSend::sendS(const boost::uint16_t &i, int flag) {
	sendBuf((void*) &i, sizeof(i), flag);
}

void DataSend::sendI(const boost::int32_t &i, int flag) {
	sendBuf((void*) &i, sizeof(i), flag);
}

void DataSend::sendL(const boost::int64_t &i, int flag) {
	sendBuf((void*) &i, sizeof(i), flag);
}

void DataSend::send(const string &str, char flag) {
	sendBuf((void*) str.c_str(), str.length(), flag);
}

//void DataSend::sendV(const vector<std::string> &array, char flag) {
//	boost::uint16_t size = (boost::uint16_t) array.size();
//	if(size <= 0){
//		sendS(size, flag);
//		return;
//	}
//	sendS(size, ZMQ_SNDMORE);
//	boost::uint16_t i = 0;
//	for (;i < size-1; i++) {
//		send(array[i], ZMQ_SNDMORE);
//	}
//	send(array[i], flag);
//}

void DataSend::sendBuf(const void *buf, unsigned long len, int flags) {
	if(isDrop) {
		if((flags & ZMQ_SNDMORE) == 0) {
			isDrop = false;
		}
		return;
	}

	if(len >= MAX_TEXT) {
		len = MAX_TEXT - 1;
	}

	if(zmq_send(sock, buf,len, flags) < 0) {
		int error = zmq_errno();
		if(error == EAGAIN){
			cout << "\n[ \033[33mWARNING\033[0m ] sendBuf-error:try_again(reach HWM's top). drop message." << endl;
			isDrop = true;
		} else {
			zmq_disconnect(sock, this->queueName.c_str());
			zmq_connect(sock, this->queueName.c_str());
			cout << "\n[ \033[31mERROR\033[0m ] zmq send error:" << error << endl;
		}
	}
}

} /* namespace base */
} /* namespace common */
